package repository;

import com.alibaba.fastjson.JSONObject;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.net.URISyntaxException;

public class HBaseRepository {
    private Logger logger = LoggerFactory.getLogger(HBaseRepository.class);
    private Configuration configuration;
    private Admin admin;
    private Connection connection;
    private Table table;
    private static String TABLE_NAME = "lian_jia_ershoufang";
    byte[] columnFamily = Bytes.toBytes("ershoufang");

    public HBaseRepository() throws URISyntaxException, IOException {
        configuration = HBaseConfiguration.create();
        configuration.addResource(new Path(ClassLoader.getSystemResource("hbase/hbase-site.xml").toURI()));
        configuration.addResource(new Path(ClassLoader.getSystemResource("hbase/core-site.xml").toURI()));
        connection = ConnectionFactory.createConnection(configuration);
        admin = connection.getAdmin();
        createTable();
        table = connection.getTable(TableName.valueOf(TABLE_NAME));
    }

    private void createTable() throws IOException {
        TableName tableName = TableName.valueOf(TABLE_NAME);
        if(admin.tableExists(tableName)){
            return ;
        }
        TableDescriptorBuilder tableDescriptorBuilder = TableDescriptorBuilder.newBuilder(tableName);
        ColumnFamilyDescriptor columnFamilyDescriptor = ColumnFamilyDescriptorBuilder.of("ershoufang");
        tableDescriptorBuilder.setColumnFamily(columnFamilyDescriptor);
        TableDescriptor tableDescriptor = tableDescriptorBuilder.build();
        admin.createTable(tableDescriptor);
    }

    public void putColumn(JSONObject row) throws IOException {
        Put put = new Put(Bytes.toBytes(row.getString("request_id")));
        row.keySet().stream().forEach(key -> {
            byte[] value = Bytes.toBytes(row.getString(key));
            put.addColumn(columnFamily,Bytes.toBytes(key),value);
        });
        table.put(put);
        logger.info("成功保存一条数据。。。。。");
    }

}
